package log;


import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;


import java.io.File;
import java.io.FilenameFilter;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.TimerTask;
import java.util.UUID;
/**
 * 1.获取源文件
 * 2.从源文件过滤出我们的数据  后缀是 .log
 * 3.把过滤出来的文件 移动至 待上传文件夹
 * 4. 创建出来hdfs的连接
 * 5.判断hdfs 里面的文件夹是否创建  没有则创建
 * 6.遍历 待上传文件夹  把 文件上传至hdfs 注意 前缀 后缀
 * 7.上传成功后  把文件移动至 备份文件夹中
 *
 *
 *
 *
 *
 *
 */
public class log extends TimerTask {

    public void run() {
        try {
            //获取本次采集时的日期
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH");
            String day = sdf.format(new Date()).replace("-", "");
            System.out.println(day);
            //获取源文件(source File)
            File sourceFile = new File("f:\\log\\app_log");
            //筛选后缀为.log的文件
            // FilenameFilter()根据名字筛选
            File[] files = sourceFile.listFiles(new FilenameFilter() {
                public boolean accept(File dir, String name) {
                    //判断文件后缀是否为.log是返回true不是返回false
                    if(name.endsWith(".log")){
                        return true;
                    }
                        return false;
                }
            });
        //显示需要采集的文件
            System.out.println("需要采集的文件:"+ Arrays.toString(files));
        //将需要采集的文件上传到临时上传目录
        File filesStay = new File("f:\\log\\stay");
        for(File file1 : files){
            FileUtils.moveFileToDirectory(file1,filesStay,true);
        }
        //显示上传成功的文件
            System.out.println("上述文件移动到了上传目录"+filesStay.getAbsolutePath());
        //构造一个Hdfs客户端对象
            FileSystem fs = FileSystem.get(new URI("hdfs://192.168.88.5:9000"),new Configuration(),"root");

        //检查HDFS中的日期目录是否存在，如果不存在，则创建
            Path DestPath = new Path("/logger/"+day);
            if(!fs.exists(DestPath)){
                fs.mkdirs(DestPath);
            }
        //检查本地备份目录是否存在，如果不在则创建
           File backupDir = new File("f:/log/backups/"+ day + "/");
            if(!backupDir.exists()){
                backupDir.mkdir();
            }
            File[] toUploadFiles = filesStay.listFiles();
            for(File file2 : toUploadFiles){
        //上传文件到HDFS并改名jie_log_
            Path destPath = new Path(DestPath + "/" + "jie_log_" + UUID.randomUUID() + ".log");
            fs.copyFromLocalFile(new Path(file2.getAbsolutePath()),destPath);
        //显示上传文件完成
            System.out.println("上传文件完成:" + file2.getAbsolutePath() + "-->" + destPath);
        //将传输完成的文件移动到备份目录
            FileUtils.moveFileToDirectory(file2,backupDir,true);
        //显示备份完成
            System.out.println("文件备份完成" + file2.getAbsolutePath() + "-->" + backupDir);
            }
        }catch(Exception e){
            e.fillInStackTrace();
        }

    }
}
